package org.budo.canal.logposition.mixed;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager;
import com.alibaba.otter.canal.protocol.position.LogPosition;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
@Getter
@Setter
public abstract class AbstractPeriodMixedLogPositionManager extends AbstractLogPositionManager {
    private MemoryLogPositionManager memoryLogPositionManager;

    private Long period;

    private Set<String> persistDestinationTasks;

    private ScheduledExecutorService executorService;

    private Boolean started = false;

    public AbstractPeriodMixedLogPositionManager() {
        this.memoryLogPositionManager = new MemoryLogPositionManager();
        this.period = 1000L;

        this.persistDestinationTasks = Collections.synchronizedSet(new HashSet<String>());
        this.executorService = Executors.newScheduledThreadPool(1);
    }

    /**
     * 读时内存中有就用内存的,没有就读ZK
     */
    @Override
    public LogPosition getLatestIndexBy(String destination) {
        LogPosition fromMem = this.getMemoryLogPositionManager().getLatestIndexBy(destination);

        if (null != fromMem && null != fromMem.getPostion() && null != fromMem.getIdentity()) {
            return fromMem;
        }

        LogPosition fromZk = this.getLatestIndexBy_0(destination);
        log.warn("#72 destination=" + destination + ", fromMem=" + fromMem + ", fromZk=" + fromZk);

        return fromZk;
    }

    @Override
    public void start() {
        if (this.started) {
            log.error("#66 started, this=" + this);
        }
        this.started = true;

        super.start();

        this.getMemoryLogPositionManager().start();

        // 启动定时工作任务
        Runnable runnable = new Runnable() {
            public void run() {
                Set<String> persistDestinationTasks = AbstractPeriodMixedLogPositionManager.this.getPersistDestinationTasks();
                List<String> destinations = new ArrayList<String>(persistDestinationTasks);
                for (String destination : destinations) {
                    try { // 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次
                        LogPosition logPosition = AbstractPeriodMixedLogPositionManager.this.getLatestIndexBy(destination);
                        AbstractPeriodMixedLogPositionManager.this.persistLogPosition_0(destination, logPosition); // 写到ZK
                        persistDestinationTasks.remove(destination);
                    } catch (Throwable e) {
                        log.error("#88 period update" + destination + " curosr failed!", e);
                    }
                }
            }
        };

        this.getExecutorService().scheduleAtFixedRate(runnable, this.getPeriod(), this.getPeriod(), TimeUnit.MILLISECONDS);
    }

    /**
     * 写时先写内存,然后异步写到ZK
     */
    @Override
    public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
        this.getPersistDestinationTasks().add(destination);

        this.getMemoryLogPositionManager().persistLogPosition(destination, logPosition);
    }

    @Override
    public void stop() {
        super.stop();

        this.getMemoryLogPositionManager().stop();

        this.getExecutorService().shutdown();
    }

    protected abstract void persistLogPosition_0(String destination, LogPosition logPosition);

    protected abstract LogPosition getLatestIndexBy_0(String destination);
}